package cn.wangjie.spark.format

import org.apache.hadoop.io.{IntWritable, Text}
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat
import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.rdd.RDD

/**
 * 保存结果RDD到文件，采用OutputFormat方式
 */
object SparkOutputFormatTest {
	
	def main(args: Array[String]): Unit = {
		
		// 构建SparkContext上下文实例对象
		val sc: SparkContext = {
			// 1.a 创建SparkConf对象，设置应用属性，比如应用名称和运行模式
			val sparkConf = new SparkConf()
				.setAppName(this.getClass.getSimpleName.stripSuffix("$"))
				.setMaster("local[2]")
			// 1.b 创建实例对象
			SparkContext.getOrCreate(sparkConf)
		}
		
		// =====================================================================
		// TODO: 1. 从本地文件系统读取数据
		val inputRDD: RDD[String] = sc.textFile("datas/wordcount/wordcount.data")
		
		// TODO: 2. 分析数据，调用RDD中函数
		val resultRDD: RDD[(Text, IntWritable)] = inputRDD
			// 过滤不合格的数据
			.filter(line => null != line && line.trim.length > 0)
			// 每行数据分割为单词
			.flatMap(line => line.trim.split("\\s+"))
			// 每个单词出现一次
			.mapPartitions{iter => iter.map(word => (word, 1))}
			// 分组，聚合统计
			.reduceByKey((tmp, item) => tmp + item)
			// 最好转换为指定格式
			.map{case (word, count) =>
				new Text(word) -> new IntWritable(count)
			}
		
		// TODO: 3. 输出结果数据，调用RDD中Action函数
		/*
		  def saveAsNewAPIHadoopFile(
		      path: String,
		      keyClass: Class[_],
		      valueClass: Class[_],
		      outputFormatClass: Class[_ <: NewOutputFormat[_, _]],
		      conf: Configuration = self.context.hadoopConfiguration
		  ): Unit
		 */
		resultRDD.saveAsNewAPIHadoopFile(
			"datas/spark/wc-output/", //
			classOf[Text], //
			classOf[IntWritable], //
			classOf[TextOutputFormat[Text, IntWritable]]
		)
		
		
		// 应用结束，关闭资源
		sc.stop()
	}
	
}
